/**
* Copyright (C) 2008 Happy Fish / YuQing
*
* FastDFS may be copied only under the terms of the GNU General
* Public License V3, which may be found in the FastDFS source kit.
* Please visit the FastDFS Home Page http://www.csource.org/ for more detail.
**/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <unistd.h>
#include <string.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <pthread.h>
#include <event.h>
#include "shared_func.h"
#include "logger.h"
#include "hash.h"
#include "global.h"
#include "global.h"
#include "ini_file_reader.h"
#include "sockopt.h"
#include "fnio_proto.h"
#include "client_io.h"
#include "server_io.h"
#include "fdht_proto_types.h"

static int64_t session_id_counter = 0;

/* key is client's session id and value is (struct fnio_client_info *) */
static HashArray client_hash;

static void server_sock_read(int sock, short event, void *arg);
static void client_sock_read_data(int sock, short event, void *arg);
static void client_sock_write_data(int sock, short event, void *arg);
static struct event ev_sock_server;

int client_io_init(int server_sock)
{
	int result;

	if ((result=hash_init(&client_hash, PJWHash, 200, 0.5)) != 0)
	{
		return result;
	}

	event_set(&ev_sock_server, server_sock, EV_READ | EV_PERSIST, \
		server_sock_read, &ev_sock_server);
	if ((result=event_base_set(g_event_base, &ev_sock_server)) != 0)
	{
		logCrit("file: "__FILE__", line: %d, " \
			"event_base_set fail.", __LINE__);
		return result;
	}
	if ((result=event_add(&ev_sock_server, NULL)) != 0)
	{
		logCrit("file: "__FILE__", line: %d, " \
			"event_add fail.", __LINE__);
		return result;
	}

	return 0;
}

static void server_sock_read(int sock, short event, void *arg)
{
	int incomesock;
	struct sockaddr_in inaddr;
	unsigned int sockaddr_len;
	struct fnio_client_info *pClientInfo;
	int result;

	sockaddr_len = sizeof(inaddr);
	incomesock = accept(sock, (struct sockaddr*)&inaddr, &sockaddr_len);

	if (incomesock < 0) //error
	{
		logError("file: "__FILE__", line: %d, " \
			"accept failed, " \
			"errno: %d, error info: %s", \
			__LINE__, errno, strerror(errno));
		return;
	}
		
	pClientInfo = NULL;
	do
	{
	if (tcpsetnonblockopt(incomesock, g_network_timeout) != 0)
	{
		break;
	}

	pClientInfo = (struct fnio_client_info *)malloc( \
			sizeof(struct fnio_client_info));
	if (pClientInfo == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc %d bytes fail, " \
			"errno: %d, error info: %s", \
			__LINE__, sizeof(struct fnio_client_info), \
			errno, strerror(errno));
		break;
	}
	memset(pClientInfo, 0, sizeof(struct fnio_client_info));
	pClientInfo->client_sock = -1;
	pClientInfo->last_server_conn_errno = -1;

	getIpaddr(getpeername, incomesock, pClientInfo->client_ip, \
		sizeof(pClientInfo->client_ip));

	pClientInfo->pRecvNode = free_queue_pop();
	if (pClientInfo->pRecvNode == NULL)
	{
		logError("file: "__FILE__", line: %d, " \
			"malloc task buff failed", \
			__LINE__);

		break;
	}

	if ((result=fnio_malloc_send_buff(&(pClientInfo->sbuff))) != 0)
	{
		break;
	}

	pClientInfo->client_sock = incomesock;
	pClientInfo->session_id = ++session_id_counter;

	logInfo("pClientInfo->session_id==%lld", pClientInfo->session_id);

	event_set(&pClientInfo->ev_read, incomesock, EV_READ, \
		client_sock_read_data, pClientInfo);
	if ((result=event_base_set(g_event_base, \
			&pClientInfo->ev_read)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"event_base_set fail.", __LINE__);
		break;
	}
	if ((result=event_add(&pClientInfo->ev_read, &g_network_tv)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"event_add fail.", __LINE__);
		break;
	}

	event_set(&pClientInfo->ev_write, incomesock, EV_WRITE, \
		client_sock_write_data, pClientInfo);
	if ((result=event_base_set(g_event_base, \
			&pClientInfo->ev_write)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"event_base_set fail.", __LINE__);
		break;
	}

	if (hash_insert(&client_hash, &pClientInfo->session_id, \
			sizeof(pClientInfo->session_id), pClientInfo) < 0)
	{
		break;
	}

	return;
	} while (0);

	close(incomesock);
	if (pClientInfo != NULL)
	{
		if (pClientInfo->pRecvNode != NULL)
		{
			free_queue_push(pClientInfo->pRecvNode);
		}
		if (pClientInfo->sbuff.send_buff != NULL)
		{
			free(pClientInfo->sbuff.send_buff);
		}
		free(pClientInfo);
	}
}

static void destroy_client_info(struct fnio_client_info *pClientInfo)
{
	event_del(&pClientInfo->ev_read);
	if (pClientInfo->client_sock >= 0)
	{
		close(pClientInfo->client_sock);
	}
	free_queue_push(pClientInfo->pRecvNode);
	free(pClientInfo->sbuff.send_buff);

	hash_delete(&client_hash, &pClientInfo->session_id, \
			sizeof(pClientInfo->session_id));
	free(pClientInfo);
}

static void client_sock_read_data(int sock, short event, void *arg)
{
	struct fnio_client_info *pClientInfo;
	int bytes;
	int recv_bytes;
	int result;
	int header_size;

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		header_size = sizeof(FDHTProtoHeader);
	}
	else
	{
		header_size = sizeof(FNIOProtoHeader);
	}

	pClientInfo = (struct fnio_client_info *)arg;
	if (event == EV_TIMEOUT)
	{
		if (pClientInfo->pRecvNode->offset > 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"recv from client %s timeout, " \
				"recv bytes: %d, expect length: %d", \
				__LINE__, pClientInfo->client_ip, \
				pClientInfo->pRecvNode->offset, \
				pClientInfo->pRecvNode->length);
			return;
		}

		if ((result=event_add(&pClientInfo->ev_read, &g_network_tv)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"event_add fail.", __LINE__);
		}
		return;
	}

	if (pClientInfo->pRecvNode->length == 0) //recv header
	{
		recv_bytes = header_size - pClientInfo->pRecvNode->offset;
	}
	else
	{
		recv_bytes = pClientInfo->pRecvNode->length - \
				pClientInfo->pRecvNode->offset;
	}

	do
	{
	if (pClientInfo->pRecvNode->offset + recv_bytes > pClientInfo->pRecvNode->size)
	{
		char *pTemp;

		pTemp = pClientInfo->pRecvNode->data;
		pClientInfo->pRecvNode->data = (char *)realloc( \
				pClientInfo->pRecvNode->data, \
				pClientInfo->pRecvNode->size + recv_bytes);
		if (pClientInfo->pRecvNode->data == NULL)
		{
			logError("file: "__FILE__", line: %d, " \
				"malloc %d bytes failed, " \
				"errno: %d, error info: %s", __LINE__, \
				pClientInfo->pRecvNode->size + recv_bytes, \
				errno, strerror(errno));

			pClientInfo->pRecvNode->data = pTemp;  //restore old data
			break;
		}

		pClientInfo->pRecvNode->size += recv_bytes;
	}

	//printf("client read data..................\n");
	bytes = recv(sock, pClientInfo->pRecvNode->data + \
			pClientInfo->pRecvNode->offset, recv_bytes, 0);
	//printf("client read data, bytes=%d\n", bytes);
	if (bytes < 0)
	{
		if (!(errno == EAGAIN || errno == EWOULDBLOCK))
		{
			logError("file: "__FILE__", line: %d, " \
				"recv from client %s failed, " \
				"errno: %d, error info: %s", \
				__LINE__, pClientInfo->client_ip, \
				errno, strerror(errno));
			break;
		}

		if ((result=event_add(&pClientInfo->ev_read, &g_network_tv)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"event_add fail.", __LINE__);
			break;
		}
		return;
	}
	else if (bytes == 0)
	{
		break;
	}

	if (pClientInfo->pRecvNode->length == 0) //header
	{
		if (pClientInfo->pRecvNode->offset+bytes < header_size)
		{
			pClientInfo->pRecvNode->offset += bytes;
			if ((result=event_add(&pClientInfo->ev_read, \
					&g_network_tv)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"event_add fail.", __LINE__);
			}
			return;
		}

		if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
		{
		pClientInfo->pRecvNode->length = buff2int(((FDHTProtoHeader *) \
					pClientInfo->pRecvNode->data)->pkg_len);
		}
		else
		{
		pClientInfo->pRecvNode->length = buff2int(((FNIOProtoHeader *) \
					pClientInfo->pRecvNode->data)->pkg_len);
		}

		if (pClientInfo->pRecvNode->length < 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, pkg length: %d < 0", \
				__LINE__, pClientInfo->client_ip, \
				pClientInfo->pRecvNode->length);

			break;
		}

		pClientInfo->pRecvNode->length += header_size;
		if (pClientInfo->pRecvNode->length > g_max_buff_size)
		{
			logError("file: "__FILE__", line: %d, " \
				"client ip: %s, pkg length: %d > max pkg " \
				"size: %d", __LINE__, pClientInfo->client_ip, \
				pClientInfo->pRecvNode->length, g_max_buff_size);

			break;
		}
	}

	pClientInfo->pRecvNode->offset += bytes;
	if (pClientInfo->pRecvNode->offset >= pClientInfo->pRecvNode->length) //recv one pkg done
	{
		struct fnio_server_info *pServerInfo;
		int key_hash_code;

		if (g_server_mode == FNIO_SERVER_MODE_SERVER && \
			((FNIOProtoHeader *)pClientInfo->pRecvNode->data)->source \
				== FNIO_PROTO_SOURCE_PROXY)
		{
			key_hash_code = buff2int(pClientInfo->pRecvNode->data +
						 sizeof(FNIOProtoHeader));

			fnio_check_server_connected_by_hash_code(key_hash_code);
		}
		else
		{
			if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
			{
				key_hash_code = buff2int(((FDHTProtoHeader *) \
				pClientInfo->pRecvNode->data)->key_hash_code);
			}
			else
			{
				key_hash_code = buff2int(((FDHTProtoHeader *) \
				(pClientInfo->pRecvNode->data + \
				sizeof(FNIOProtoHeader)))->key_hash_code);
			}

			pServerInfo = fnio_get_server(key_hash_code);
			if (pServerInfo != NULL)
			{
				if (fnio_push_to_server_send_buff( \
					pServerInfo, pClientInfo) != 0)
				{
					break;
				}
			}
			else
			{
				break;
			}
		}

		pClientInfo->pRecvNode->length = 0;
		pClientInfo->pRecvNode->offset = 0;
	}

	if ((result=event_add(&pClientInfo->ev_read, &g_network_tv)) != 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"event_add fail.", __LINE__);
	}
	return;
	} while (0);

	destroy_client_info(pClientInfo);
}

static int turn_on_client_sending(struct fnio_client_info *pClientInfo)
{
	int result;

	if (pClientInfo->sending)
	{
		return 0;
	}

	if ((result=event_add(&pClientInfo->ev_write, \
			&g_network_tv)) != 0)
	{
		destroy_client_info(pClientInfo);

		logError("file: "__FILE__", line: %d, "\
			"event_add fail, errno: %d, " \
			"error info: %s", __LINE__, \
			result, strerror(result));
	}
	else
	{
		pClientInfo->sending = true;
	}

	return result;
}

int fnio_push_to_client_send_buff(struct fnio_server_info *pServerInfo)
{
	int result;
	FNIOProtoHeader header;
	struct fnio_client_info *pClientInfo;
	FNIOProtoHeader *pHeader;
	int64_t session_id;

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		pHeader = (FNIOProtoHeader *)pServerInfo->pRecvNode->data;
		session_id = buff2long(pHeader->session_id);
	}
	else
	{
		session_id = buff2long(pServerInfo->server_session_id);
	}

	pClientInfo = (struct fnio_client_info *)hash_find(&client_hash, \
		&session_id, sizeof(session_id));
	if (pClientInfo == NULL)
	{
		logWarning("file: "__FILE__", line: %d, " \
			"session_id: "INT64_PRINTF_FORMAT" not found", \
			__LINE__, session_id);
		return ENOENT;
	}

	if (pClientInfo->session_id != session_id)
	{
		logError("file: "__FILE__", line: %d, " \
			"client session id: "INT64_PRINTF_FORMAT" != " \
			"session_id: "INT64_PRINTF_FORMAT, \
			__LINE__, pClientInfo->session_id, session_id);
	}

	if (g_server_mode == FNIO_SERVER_MODE_CLIENT)
	{
		result = fnio_push_to_send_buff(&pClientInfo->sbuff, NULL, 0, \
			pServerInfo->pRecvNode->data + sizeof(FNIOProtoHeader),
			pServerInfo->pRecvNode->length - sizeof(FNIOProtoHeader));
	}
	else
        {
		memcpy(header.session_id, pServerInfo->client_session_id, 8);
		int2buff(pServerInfo->pRecvNode->length, header.pkg_len);
		header.source = FNIO_PROTO_SOURCE_DATA;
                result = fnio_push_to_send_buff(&pClientInfo->sbuff,
                        (char *)&header, sizeof(FNIOProtoHeader),
                        pServerInfo->pRecvNode->data, pServerInfo->pRecvNode->length);
	}

	if (result != 0)
	{
		return result;
	}

	return turn_on_client_sending(pClientInfo);
}

static void client_sock_write_data(int sock, short event, void *arg)
{
	struct fnio_client_info *pClient;
	int result;
	int bytes;
	int send_bytes;

	pClient = (struct fnio_client_info *)arg;
	if (event == EV_TIMEOUT)
	{
		logError("file: "__FILE__", line: %d, " \
			"write to client %s timeout, " \
			"session id="INT64_PRINTF_FORMAT, \
			__LINE__, pClient->client_ip, pClient->session_id);

		return;
	}

	if (pClient->sbuff.send_bytes == 0)
	{
		return;
	}

	if (pClient->sbuff.send_start + pClient->sbuff.send_bytes <=  pClient->sbuff.buff_end)
	{
		send_bytes = pClient->sbuff.send_bytes;
	}
	else
	{
		send_bytes = pClient->sbuff.buff_end - pClient->sbuff.send_start;
	}

	bytes = send(sock, pClient->sbuff.send_start, send_bytes, 0);
	if (bytes < 0)
	{
		if (errno == EAGAIN || errno == EWOULDBLOCK)
		{
			if ((result=event_add(&pClient->ev_write, \
					&g_network_tv)) != 0)
			{
				logError("file: "__FILE__", line: %d, " \
					"event_add fail, " \
					"errno: %d, error info: %s", \
					__LINE__, result, strerror(result));
				destroy_client_info(pClient);
			}

			return;
		}

		logError("file: "__FILE__", line: %d, " \
			"send to client %s fail, errno:%d, error info: %s", \
			__LINE__, pClient->client_ip, errno, strerror(errno));
		destroy_client_info(pClient);
		return;
	}
	else if (bytes == 0)
	{
		logError("file: "__FILE__", line: %d, " \
			"send to client %s fail, connection disconnected", \
			__LINE__, pClient->client_ip);

		destroy_client_info(pClient);
		return;
	}

	pClient->sbuff.send_bytes -= bytes;
	pClient->sbuff.send_start += bytes;
	if (pClient->sbuff.send_start >= pClient->sbuff.buff_end)
	{
		pClient->sbuff.send_start = pClient->sbuff.send_buff;
	}

	if (pClient->sbuff.send_bytes > 0)
	{
		if ((result=event_add(&pClient->ev_write, &g_network_tv)) != 0)
		{
			logError("file: "__FILE__", line: %d, " \
				"event_add fail, " \
				"errno: %d, error info: %s", \
				__LINE__, result, strerror(result));

			destroy_client_info(pClient);
		}
	}
	else
	{
		pClient->sending = false;
	}
}

static int notify_proxy_client_conn_info(const int index, \
		const HashData *data, void *args)
{
	struct fnio_client_info *pClient;
	struct fnio_server_info *pServer;

	pServer = (struct fnio_server_info *)args;
	pClient = (struct fnio_client_info *)data->value;
	if (pClient->last_server_conn_errno != pServer->last_conn_errno)
	{
		if (fnio_push_to_send_buff(&pClient->sbuff, NULL, 0, \
				pServer->pSendNode->data, \
				pServer->pSendNode->length) == 0)
		{
			turn_on_client_sending(pClient);
			pClient->last_server_conn_errno = pServer->last_conn_errno;
		}
	}

	return 0;
}

int fnio_notify_server_conn_status(struct fnio_server_info *pServer)
{
	FNIOProtoHeader *pHeader;
	char buff[sizeof(FNIOProtoHeader) + 1];
	char *saved_buff;
	int saved_len;
	int result;

	saved_buff = pServer->pSendNode->data;
	saved_len = pServer->pSendNode->length;

	pServer->pSendNode->data = buff;
	pServer->pSendNode->length = sizeof(FNIOProtoHeader) + 1;
	memset(buff, 0, sizeof(buff));
	pHeader = (FNIOProtoHeader *)buff;
	int2buff(1, pHeader->pkg_len);
	pHeader->source = FNIO_PROTO_SOURCE_PROXY;
	*(pServer->pSendNode->data + sizeof(FNIOProtoHeader)) = \
			pServer->last_conn_errno;

	//logInfo("fnio_notify_server_conn_status, pServer->last_conn_errno=%d", pServer->last_conn_errno);

	result = hash_walk(&client_hash, notify_proxy_client_conn_info, pServer);

	pServer->pSendNode->data = saved_buff;
	pServer->pSendNode->length = saved_len;

	return result;
}

